package com.nucarf.yc.app

import com.nucarf.yc.bean.OrderSum
import com.nucarf.yc.utils.KafkaUtil
import net.minidev.json.JSONObject
import net.minidev.json.parser.JSONParser
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import scala.collection.immutable.HashMap
import scala.collection.mutable
import scala.collection.mutable._

object ConsKafka {


  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /* 设置并行度 */
    env.setParallelism(1)

    /**
     * {
    "amount": 1352.46,
    "change_time": 0,
    "op_type": 1,
    "trade_no": 5854395,
    "trade_time": 1632649483,
    "update_time": 0
}*/
    val inputStream: DataStream[String] = env.addSource( new FlinkKafkaConsumer[String](
        KafkaUtil.CONSUME_TEST,
        new SimpleStringSchema(),
        KafkaUtil.kafkaConsumerProperties()) )


    inputStream.flatMap(new OrderSumFlatMap)

    env.execute("OrderSum : ")
  }
}

class OrderSumFlatMap extends RichFlatMapFunction[String, HashMap[String, OrderSum]]{


  var orderMap: HashMap[String, OrderSum] = new HashMap[String, OrderSum]
  var idsMap: HashMap[String, HashSet[String]] = new HashMap[String, HashSet[String]]

  override def flatMap(json: String, collector: Collector[HashMap[String, OrderSum]]): Unit = {

    val parser = new JSONParser(1)
    val jsonObject: JSONObject = parser.parse(json).asInstanceOf[JSONObject]
    val trade_no: String = jsonObject.get("trade_no").toString
    val amount: String = jsonObject.get("amount").toString
    val money: BigDecimal = BigDecimal.apply(amount)

    /* 1:正常 2:撤单 3：红冲数据 */
    val op_type: String = jsonObject.get("op_type").toString
    val trade_time: Long = jsonObject.get("trade_time").toString.toLong
    val mouth: String = getMouth(trade_time)
    val update_time: Long = jsonObject.get("update_time").toString.toLong
    val change_time: Long = jsonObject.get("change_time").toString.toLong
    val sum: OrderSum = orderMap.getOrElse(mouth, null)

    if (sum == null){
      val ids = new mutable.HashSet[String]
      ids.add(trade_no)
      val idsSet: mutable.HashSet[String] = idsMap.getOrElse(mouth, ids)
      val orderSum = new OrderSum(mouth, money, idsSet.size)
      orderMap+=((mouth, orderSum))
    } else {
      val ids = new mutable.HashSet[String]
      ids.add(trade_no)
      val idsSet: mutable.HashSet[String] = idsMap.getOrElse(mouth, ids)
      val orderSum: OrderSum = get2OrderSum(sum, new OrderSum(mouth, money, idsSet.size))
      orderMap+=((mouth, orderSum))
    }

    println(orderMap)

  }

  def getMouth(time: Long): String ={
    new SimpleDateFormat("yyyy-MM").format(time * 1000)
  }

  def get2OrderSum(o1: OrderSum, o2: OrderSum): OrderSum ={
    val sum: OrderSum = o1.copy(o1.mouth,
      o1.money_total + o2.money_total,
      o1.order_id_total + o2.order_id_total)
    sum
  }
}
